package io.plus.engine.client;

import io.plus.engine.pulse.*;
import io.plus.engine.utils.lock.ReadWriteLockSet;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import tool.plus.$;
import tool.plus.Clock;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;

@Slf4j
@Data
public class Client {
    private ClientConfiguration clientConfiguration;
    private AsynchronousChannelGroup asynchronousChannelGroup;

    public Client(final ClientConfiguration clientConfiguration) throws IOException {
        super();
        this.clientConfiguration = clientConfiguration;
        this.asynchronousChannelGroup = AsynchronousChannelGroup.withThreadPool(clientConfiguration.groupThreadPoolExecutor);
        heartbeat();
        autoConnect();
    }

    public void asynConnect(Node serverNode) throws Exception {
        asynConnect(serverNode, null);
    }

    public void asynConnect(Node serverNode, Integer timeout) throws Exception {
        asynConnect(serverNode, serverNode.getIp(), serverNode.getPort(), timeout);
    }

    public void asynConnect(Node serverNode, String bindIp, Integer bindPort, Integer timeout) throws Exception {
        connect(serverNode, bindIp, bindPort, null, timeout, false);
    }

    public ClientAsynchronousSocketChannelContext connect(Node serverNode) throws Exception {
        return connect(serverNode, null);
    }

    public ClientAsynchronousSocketChannelContext connect(Node serverNode, Integer timeout) throws Exception {
        return connect(serverNode, serverNode.getIp(), serverNode.getPort(), timeout);
    }

    /**
     * @param serverNode
     * @param bindIp
     * @param bindPort
     * @param initClientChannelContext
     * @param timeout                  超时时间，单位秒
     */
    public ClientAsynchronousSocketChannelContext connect(Node serverNode, String bindIp, Integer bindPort, ClientAsynchronousSocketChannelContext initClientChannelContext, Integer timeout) throws Exception {
        return connect(serverNode, bindIp, bindPort, initClientChannelContext, timeout, true);
    }

    /**
     * @param serverNode
     * @param bindIp
     * @param bindPort
     * @param initClientChannelContext
     * @param timeout                  超时时间，单位秒
     * @param isSynchronized           true: 同步, false: 异步
     */
    private ClientAsynchronousSocketChannelContext connect(Node serverNode, String bindIp, Integer bindPort, ClientAsynchronousSocketChannelContext initClientChannelContext, Integer timeout, boolean isSynchronized)
            throws Exception {
        AsynchronousSocketChannel asynchronousSocketChannel = null;
        ClientAsynchronousSocketChannelContext channelContext = null;
        boolean isReconnect = initClientChannelContext != null;

        long start = Clock.currentTimeMillis;
        asynchronousSocketChannel = AsynchronousSocketChannel.open(asynchronousChannelGroup);
        long end = Clock.currentTimeMillis;
        long interval = end - start;
        if (interval >= 100) {
            log.error("{}, open 耗时:{} ms", channelContext, interval);
        }

        asynchronousSocketChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
        asynchronousSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
        asynchronousSocketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);

        //
        InetSocketAddress bind = null;
        if (bindPort != null && bindPort > 0) {
            if ($.isNotNullAndNotEmpty(bindIp)) {
                bind = new InetSocketAddress(bindIp, bindPort + 11);
            } else {
                bind = new InetSocketAddress(bindPort + 11);
            }
        }
        if (bind != null) {
            asynchronousSocketChannel.bind(bind);
        }
//
        channelContext = initClientChannelContext;


        InetSocketAddress inetSocketAddress = new InetSocketAddress(serverNode.getIp(), serverNode.getPort());

        ConnectionCompletionVo attachment = new ConnectionCompletionVo(channelContext, this, isReconnect, asynchronousSocketChannel, serverNode, bindIp, bindPort);

        if (isSynchronized) {
            Integer realTimeout = timeout;
            if (realTimeout == null) {

                realTimeout = 5;
            }

            CountDownLatch countDownLatch = new CountDownLatch(1);
            attachment.setCountDownLatch(countDownLatch);
            asynchronousSocketChannel.connect(inetSocketAddress, attachment, clientConfiguration.getConnectionCompletionHandler());
            boolean await = countDownLatch.await(realTimeout, TimeUnit.SECONDS);
            if (await) {
                return attachment.getClientChannelContext();
            } else {
                log.error("countDownLatch.await(realTimeout, TimeUnit.SECONDS) 返回false ");
                return attachment.getClientChannelContext();
            }
        } else {
            asynchronousSocketChannel.connect(inetSocketAddress, attachment, clientConfiguration.getConnectionCompletionHandler());
            return null;
        }
    }

    /**
     * @param serverNode
     * @param bindIp
     * @param bindPort
     * @param timeout    超时时间，单位秒
     */
    public ClientAsynchronousSocketChannelContext connect(Node serverNode, String bindIp, Integer bindPort, Integer timeout) throws Exception {
        return connect(serverNode, bindIp, bindPort, null, timeout);
    }

    /**
     * @param channelContext
     * @param timeout        单位秒
     */
    public void reconnect(ClientAsynchronousSocketChannelContext channelContext, Integer timeout) throws Exception {
        connect(channelContext.getServerNode(), channelContext.getBindIp(), channelContext.getBindPort(), channelContext, timeout);
    }

    /**
     * 定时任务：发心跳
     */
    private void heartbeat() {
        final ClientHandler clientHandler = clientConfiguration.getClientHandler();
        final String id = clientConfiguration.getId();
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (!clientConfiguration.isStopped()) {
                    ReadWriteLockSet<AsynchronousSocketChannelContext> connecteds = clientConfiguration.connecteds;
                    ReadLock readLock = connecteds.getReadLock();
                    readLock.lock();
                    try {
                        Set<AsynchronousSocketChannelContext> asynchronousSocketChannelContextSet = connecteds.getObject();
                        long currentTimeMillis = Clock.currentTimeMillis;
                        for (AsynchronousSocketChannelContext asynchronousSocketChannelContext1 : asynchronousSocketChannelContextSet) {
                            ClientAsynchronousSocketChannelContext channelContext = (ClientAsynchronousSocketChannelContext) asynchronousSocketChannelContext1;
                            if (channelContext.isClosed || channelContext.isRemoved) {
                                continue;
                            }

                            AsynchronousSocketChannelState asynchronousSocketChannelState = channelContext.asynchronousSocketChannelState;
                            long compareTime = Math.max(asynchronousSocketChannelState.latestTimeOfReceivedByte, asynchronousSocketChannelState.latestTimeOfSentPacket);
                            long interval = currentTimeMillis - compareTime;
                            if (interval >= clientConfiguration.heartbeatTimeout / 2) {
                                PacketMetadata packetMetadata = clientHandler.heartbeatPacket(channelContext);
                                if (packetMetadata != null) {
                                    if (log.isInfoEnabled()) {
                                        log.info("{}发送心跳包", channelContext.toString());
                                    }
                                    Io.send(channelContext, packetMetadata);
                                }
                            }
                        }


                    } catch (Throwable e) {
                        log.error("", e);
                    } finally {
                        try {
                            readLock.unlock();
                            Thread.sleep(clientConfiguration.heartbeatTimeout / 4);
                        } catch (Throwable e) {
                            log.error(e.toString(), e);
                        } finally {

                        }
                    }
                }
            }
        }, "io-heartbeat" + id).start();
    }

    /**
     * 启动重连任务
     */
    private void autoConnect() {
        final ReConnectConfiguration reConnectConfiguration = clientConfiguration.getReConnectConfiguration();
        if (reConnectConfiguration == null || reConnectConfiguration.getInterval() <= 0) {
            return;
        }

        final String id = clientConfiguration.getId();
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!clientConfiguration.isStopped()) {
                    log.error("closeds:{}, connections:{}", clientConfiguration.closeds.size(), clientConfiguration.connections.size());
                    LinkedBlockingQueue<AsynchronousSocketChannelContext> queue = reConnectConfiguration.getQueue();
                    ClientAsynchronousSocketChannelContext channelContext = null;
                    try {
                        channelContext = (ClientAsynchronousSocketChannelContext) queue.take();
                    } catch (InterruptedException e1) {
                        log.error(e1.toString(), e1);
                    }
                    if (channelContext == null) {
                        continue;
                    }
                    //已经删除的，不需要重新再连
                    if (channelContext.isRemoved) {
                        continue;
                    }
                    long sleepTime = reConnectConfiguration.getInterval() - (Clock.currentTimeMillis - channelContext.asynchronousSocketChannelState.timeInReconnQueue);
                    if (sleepTime > 0) {
                        try {
                            Thread.sleep(sleepTime);
                        } catch (InterruptedException e) {
                            log.error(e.toString(), e);
                        }
                    }
                    //已经删除的和已经连上的，不需要重新再连
                    if (channelContext.isRemoved || !channelContext.isClosed) {
                        continue;
                    } else {
                        ReConnectTask reConnectTask = channelContext.getReConnectTask();
                        if (reConnectTask == null) {
                            synchronized (channelContext) {
                                reConnectTask = channelContext.getReConnectTask();
                                if (reConnectTask == null) {
                                    reConnectTask = new ReConnectTask(channelContext, Client.this, reConnectConfiguration.getThreadPoolExecutor());
                                    channelContext.setReConnectTask(reConnectTask);
                                }
                            }
                        }
                        reConnectTask.execute();
                    }
                }
            }
        });
        thread.setName("io-reconnect-" + id);
        thread.setDaemon(true);
        thread.start();

    }

    public boolean stop() {
        boolean ok = true;
        try {
            clientConfiguration.groupThreadPoolExecutor.shutdown();
        } catch (Exception e1) {
            log.error(e1.toString(), e1);
        }
        try {
            clientConfiguration.lockCapableTaskThreadPoolExecutor.shutdown();
        } catch (Exception e1) {
            log.error(e1.toString(), e1);
        }

        clientConfiguration.setStopped(true);
        try {
            ok = ok && clientConfiguration.groupThreadPoolExecutor.awaitTermination(6000, TimeUnit.SECONDS);
            ok = ok && clientConfiguration.lockCapableTaskThreadPoolExecutor.awaitTermination(6000, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error(e.getLocalizedMessage(), e);
        }
        log.info("client resource has released");
        return ok;
    }
}
